package com.yihaodian.cluster;

import java.util.Collection;
import java.util.concurrent.CopyOnWriteArrayList;

import org.apache.log4j.Logger;

import com.google.inject.Inject;
import com.yihaodian.cluster.discovery.NodeDiscovery;
import com.yihaodian.cluster.discovery.NodeDiscovery.InitClusterStateListener;
import com.yihaodian.cluster.discovery.NodeDiscovery.NewClusterStateListener;
import com.yihaodian.cluster.discovery.NodeDiscovery.NodeChangeListener;
import com.yihaodian.index.IndexException;

public class DefaultClusterService implements ClusterService {
	private static Logger logger = Logger
			.getLogger(DefaultClusterService.class);

	NodeDiscovery discovery;

	ClusterState current;

	CopyOnWriteArrayList<ClusterStateListener> clusterStateListeners = new CopyOnWriteArrayList<ClusterStateListener>();

	ClusterStateLocal clusterStateLocal;

	@Inject
	public DefaultClusterService(NodeDiscovery nodeDiscovery,
			ClusterStateLocal clusterStateLocal) {
		this.discovery = nodeDiscovery;
		this.clusterStateLocal = clusterStateLocal;
	}

	/**
	 * listening ClusterState change from master
	 */
	class MasterClusterStateListener implements NewClusterStateListener {

		@Override
		public void onNewClusterState(ClusterState state) {
			handleClusterState("update from master ", state);
		}
	}

	/**
	 * init state, master would init.
	 * 
	 */
	class InitStateListener implements InitClusterStateListener {

		@Override
		public void initClusterState(ClusterState state) {
			ClusterStateUpdate update = new ClusterStateUpdate(
					ClusterStateUpdateType.cluster_state_init);

			update.setState(state);
			updateClusterState(update);
		}
	}

	/**
	 * handle node change, master handle it.
	 * 
	 */
	class NodeListener implements NodeChangeListener {

		@Override
		public void onNodeEvent(Collection<Node> nodes) {
			ClusterStateUpdate update = new ClusterStateUpdate(
					ClusterStateUpdateType.cluster_node_change);
			update.setNodeList(nodes);

			updateClusterState(update);
		}

	}

	@Override
	public Node getLocalNode() {
		return discovery.getLocalNode();
	}

	@Override
	public ClusterState state() {
		return current;
	}

	/**
	 * notify to upper listener.
	 * 
	 * @param source
	 * @param newClusterState
	 */
	void handleClusterState(String source, ClusterState newClusterState) {
		ClusterState prevClusterState = current;
		current = newClusterState;
		ClusterChangedEvent event = new ClusterChangedEvent(source,
				newClusterState, prevClusterState);
		for (ClusterStateListener listener : clusterStateListeners) {
			listener.onClusterState(event);
		}
	}

	/**
	 */
	@Override
	public void updateClusterState(ClusterStateUpdate update) {
		Node local = discovery.getLocalNode();

		ClusterState newClusterState = null;
		ClusterStateUpdateType type = update.getType();

		if (type == ClusterStateUpdateType.cluster_state_init) {
			update.setMasterNode(local);

			// init, compare local state with zk state
			ClusterState localState = clusterStateLocal.read();

			newClusterState = update.newClusterState(localState);
		} else {
			checkStateInit();

			newClusterState = update.newClusterState(current);
		}

		// double check master
		if (newClusterState.masterNode() != null
				&& newClusterState.masterNode().getId().equals(local.getId())) {
			logger.info(local.getId() + " publish: " + update.getType());
			discovery.publishState(newClusterState);
		}

		handleClusterState(type.name(), newClusterState);
	}

	void checkStateInit() {
		if (current == null) {
			try {
				Thread.sleep(5 * 1000);
			} catch (InterruptedException e) {
				// ignore
			}
		}
		if (current == null)
			throw new IndexException("ClusterState is not init yet");
	}

	@Override
	public void addClusterStateListener(ClusterStateListener listener) {
		clusterStateListeners.add(listener);
	}

	@Override
	public void startup() {
		// master node
		this.discovery.addListener(new InitStateListener());
		this.discovery.addListener(new NodeListener());

		// slave nodes
		this.discovery.addListener(new MasterClusterStateListener());
	}

	@Override
	public void shutdown() {
	}

}